package com.atguigu.gmall.realtime.test;

import com.atguigu.gmall.realtime.common.base.BaseSqlApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Realtime04_DorisConnectorSql extends BaseSqlApp {
    public static void main(String[] args) {
        new Realtime04_DorisConnectorSql().start(5678, 1, "Realtime04_Doris_Connector_Sql");
    }

    @Override
    public void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        // 通过sql的形式读取Doris的数据

        // doris source
        String sourceTable =
                "CREATE TABLE source_table (" +
                        "                siteid INT," +
                        "                citycode SMALLINT," +
                        "                username STRING," +
                        "                pv BIGINT" +
                        "        ) WITH (" +
                        "                'connector' = 'doris'," +
                        "                'fenodes' = 'hadoop102:7030'," +
                        "                'table.identifier' = 'test.table1'," +
                        "                'username' = 'root'," +
                        "                'password' = 'aaaaaa'" +
                        "        );";
        tableEnv.executeSql(sourceTable);

        // 读取数据
        // tableEnv.sqlQuery("select siteid, citycode, username, pv from source_table").execute().print();

        // 将查询到的结果写入到table2中
        Table resultTable = tableEnv.sqlQuery("select siteid, citycode, username, pv from source_table");
        String sinkTable =
                "CREATE TABLE sink_table (" +
                        "                siteid INT," +
                        "                citycode SMALLINT," +
                        "                username STRING," +
                        "                pv BIGINT" +
                        "        ) WITH (" +
                        "                'connector' = 'doris'," +
                        "                'fenodes' = 'hadoop102:7030'," +
                        "                'table.identifier' = 'test.table2'," +
                        "                'username' = 'root'," +
                        "                'password' = 'aaaaaa', " +
                        "                'sink.enable-2pc' = 'false'" +
                        // "                'sink.label-prefix' =  'doris_label"+ System.currentTimeMillis() +"'," +
                        // "                '' = '' " +
                        "        );";
        tableEnv.executeSql(sinkTable);

        // 写入
        resultTable.executeInsert("sink_table");

    }
}
